package com.lianxi.rocketmqmvc;

import com.alibaba.fastjson.JSON;
import com.lianxi.util.DateUtil;
import com.lianxi.util.ResourcesUtil;
import org.apache.commons.io.Charsets;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.DependsOn;
import org.springframework.core.task.TaskExecutor;

import java.util.Date;
import java.util.List;

/**
 * Description : 消费者工具类
 * Date : 2018/1/23 16:05
 *
 * @author : fy
 */
@DependsOn("springContextHolder")
public class RoleConsumer {
    private static Logger logger = LoggerFactory.getLogger("RoleConsumerUtil");

//    private static TaskExecutor taskExecutor = SpringContextHolder.getBean("taskExecutor");
    /**
     * 默认name server
     */
    private static final String DEFAULT_CONSUMER_NAMESRV_ADDR = ResourcesUtil.getRocketSrvAddr();
    /**
     * 单队列并行消费最大跨度, 默认2000
     */
    private static final Integer MAX_CONCURRENT_SPAN_SIZE = 2000;
    /**
     * 并发消费时，一次消费消息的数量
     */
    private static final Integer MAX_MSG_BATCH_SIZE = 100;
    /**
     * 每个topic 的最小线程数
     */
    private static final Integer DEFAULT_MIN_THREAD_NUM = 10;

    /**
     * 每个topic的最大线程数
     */
    private static final Integer DEFAULT_MAX_THREAD_NUM = 30;

    private static final String SUBSCRIBE_TOPIC = RocketMQConstant.ROCKETMQ_ROLE_TOPIC;

    private static final String CONSUMER_GROUP = RocketMQConstant.ROCKETMQ_ROLE_GROUP;

    private static DefaultMQPushConsumer defaultMQPushConsumer;

    private RoleConsumer() {

    }

    private static DefaultMQPushConsumer getDefaultMQPullConsumer() {
        if (defaultMQPushConsumer == null) {
            synchronized (RoleConsumer.class) {
                if (defaultMQPushConsumer == null) {
                    defaultMQPushConsumer = new DefaultMQPushConsumer(CONSUMER_GROUP,true);
                }
            }
        }
        return defaultMQPushConsumer;
    }



    /**
     * 开启消费者
     */
    public static void startConsumer() {
        defaultMQPushConsumer = getDefaultMQPullConsumer();
        defaultMQPushConsumer.setNamesrvAddr(DEFAULT_CONSUMER_NAMESRV_ADDR);
        //单队列并行消费最大跨度
        defaultMQPushConsumer.setConsumeConcurrentlyMaxSpan(MAX_CONCURRENT_SPAN_SIZE);
        //并发消费时，一次消费消息的数量，默认为1，假如修改为50，此时若有100条消息，那么会创建两个线程，每个线程分配50条消息。
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(MAX_MSG_BATCH_SIZE);
        //从消息队列头开始消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //消费方式  集群消费
        defaultMQPushConsumer.setMessageModel(MessageModel.BROADCASTING);
        //最大消费线程数
        defaultMQPushConsumer.setConsumeThreadMax(DEFAULT_MAX_THREAD_NUM);
        //最小消费线程数
        defaultMQPushConsumer.setConsumeThreadMin(DEFAULT_MIN_THREAD_NUM);
        //vip通道关闭，开启的话会走netty通道
        defaultMQPushConsumer.setVipChannelEnabled(false);
        try {
            //订阅 和日志相同的tag进行消费
            defaultMQPushConsumer.subscribe(RocketMQConstant.ROCKETMQ_COUNT_TOPIC, "*");
            defaultMQPushConsumer.registerMessageListener((List<MessageExt> list,
                                                           ConsumeConcurrentlyContext context) -> {
                list.forEach(messageExt -> {
                    String msgBody = new String(messageExt.getBody(), Charsets.UTF_8);
                    System.out.println("接收时间："+ DateUtil.formate(new Date())+" 消息内容："+msgBody);
//                    SsoCleanUserDto ssoCleanUserDto = JSON.parseObject(msgBody, SsoCleanUserDto.class);
//                    if (ssoCleanUserDto.getUserId() == -1){
//                        taskExecutor.execute(new ClearShiroTask());
//                    }else {
//                        taskExecutor.execute(new ClearShiroTask(ssoCleanUserDto.getUserId()));
//                    }
//                    logger.info(" clear  shiro  end  user id :{} " ,ssoCleanUserDto.getUserId());
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            logger.info(" --------------------- staticConsumer start ---------------------");
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    /**
     * 关闭消费者
     */
    public static void closeConsumer() {
        defaultMQPushConsumer.shutdown();
        logger.info(" ----- consumer shutdown ----- ");
    }
}
